/*
 * Copyright (c) 2020 Carnegie Lee
 * All rights reserved
 */
package com.geekc.kafka.producer;

import com.geekc.kafka.constant.KafkaConst;
import com.geekc.kafka.domain.Message;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.requestreply.RequestReplyFuture;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;

/**
 * 生产者 - Replying模式
 * @author Carnegie Lee
 */
@Component
public class ReplyingProducer {

    @Autowired(required = false)
    private ReplyingKafkaTemplate replyingKafkaTemplate;

    /**
     * 发送消息并获取响应
     * @param message 消息体
     */
    public void replyingSend(Message message) {
        ProducerRecord<String, String> producerRecord = new ProducerRecord(KafkaConst.KAFKA_TARGET_TOPIC_REPLYING, message.toString());
        RequestReplyFuture<String, String, String> replyFuture = replyingKafkaTemplate.sendAndReceive(producerRecord);
        try {
            ConsumerRecord<String, String> consumerRecord = replyFuture.get();
            System.out.println("接收到返回消息：" + consumerRecord.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
